// 文件路径: dbms/cluster/PeerNodeHandler.java
package com.dbms.cluster;

import com.dbms.DbmsCore;
import com.dbms.common.QueryResult;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;

/**
 * 在服务器端处理来自其他对等节点（Peer）连接的处理器。
 * 这是一个实现了 Runnable 接口的类，通常会在一个独立的线程中执行，
 * 以便并发处理多个来自对等节点的请求。
 */
public class PeerNodeHandler implements Runnable {
    // 与对等节点建立的Socket连接
    private final Socket peerSocket;
    // 数据库核心引擎的实例，用于执行SQL命令
    private final DbmsCore dbmsCore;

    /**
     * PeerNodeHandler 的构造函数。
     * @param socket 与对等节点建立的Socket连接。
     * @param core   当前节点的数据库核心引擎实例。
     */
    public PeerNodeHandler(Socket socket, DbmsCore core) {
        this.peerSocket = socket;
        this.dbmsCore = core;
    }

    /**
     * 线程执行的主体方法。
     */
    @Override
    public void run() {
        // 同样使用 try-with-resources 确保流和Socket在结束时被关闭
        try (
                ObjectOutputStream oos = new ObjectOutputStream(peerSocket.getOutputStream());
                ObjectInputStream ois = new ObjectInputStream(peerSocket.getInputStream())
        ) {
            // 1. 从输入流中读取并反序列化来自对等节点的请求对象。
            ClusterRequest request = (ClusterRequest) ois.readObject();

            System.out.println("从对等节点接收到转发的SQL。原始用户: " + request.getOriginalUser());

            // 2. 【核心修正】调用 DbmsCore 中一个为集群间通信设计的特定方法 `processClusteredSql`。
            // 这个方法非常重要，因为它处理的是已经被转发过的请求。
            // 它会执行SQL，但会标记这个请求为“内部集群请求”，从而防止这个请求被当前节点再次广播或转发给其他节点，
            // 避免了无限循环转发的问题。同时，它会使用原始用户名来进行权限验证。
            QueryResult result = dbmsCore.processClusteredSql(request.getSql(), request.getOriginalUser());

            // 3. 将执行后的结果（QueryResult）封装成一个 ClusterResponse 对象。
            ClusterResponse response = new ClusterResponse(result);
            // 4. 将响应对象写入输出流，发送回给发起请求的对等节点。
            oos.writeObject(response);
            oos.flush();

        } catch (Exception e) {
            // 如果在处理过程中发生任何异常，打印错误信息。
            System.err.println("处理来自 " + peerSocket.getInetAddress() + " 的对等节点请求时出错: " + e.getMessage());
            e.printStackTrace();
        }
    }
}